package streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Transform extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("DSWordCount")
  sparkConf.setMaster("local[4]")
  val sparkContext = new SparkContext(sparkConf)
  /**
   * 5s 一个窗口 即一个批次
   */
  val ssc = new StreamingContext(sparkContext, Durations.seconds(5))
  /**
   * 先在 node1 上启动 nc -lk 9999
   */
  val ds = ssc.socketTextStream("node1", 9999)
  ds.cache()
  ds.map((_, 1)).reduceByKey(_ + _).print()
  ds.transform(rdd => {
    val countRdd = rdd.map((_, 1)).reduceByKey(_ + _)
    countRdd
  }).print()
  ds.foreachRDD(rdd => {
    rdd.map((_, 1)).reduceByKey(_ + _).foreach(println)
  })

  ssc.start()
  ssc.awaitTermination()
  ssc.stop()

}
